package org.example.writebitfile;

import org.example.util.MurmurHashUtil;
import org.roaringbitmap.longlong.Roaring64Bitmap;
import org.xerial.snappy.Snappy;

import java.io.DataOutputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * {@code @projectName} bitmap17
 * {@code @Author} huzhengyang
 * {@code @Create} 2024/4/22 23:06
 */

public class SerializeToDiskWithSnappyExample {


    public static void main(String[] args) throws IOException {
        if (args.length > 0) {
            String maxLine = args[0];
            String file1 = args[1];
            String file2 = args[2];
            String snappyFile1 = args[3];
            String snappyFile2 = args[4];
            String diff1 = args[5];
            String diff2 = args[6];
            String diff3 = args[7];

            long maxLn = Long.parseLong(maxLine);
            long d1 = Long.parseLong(diff1);
            long d2 = Long.parseLong(diff2);
            long d3 = Long.parseLong(diff3);

            CountDownLatch latch = new CountDownLatch(2);

            Thread t1 = new Thread(() -> {
                try {
                    generateFile(maxLn, file1, snappyFile1, null);
                    latch.countDown();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            });

            Thread t2 = new Thread(() -> {
                try {
                    generateFile(maxLn, file2, snappyFile2, Arrays.asList(d1, d2, d3));
                    latch.countDown();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            });

            t1.start();
            t2.start();

            try {
                latch.await();
            } catch (InterruptedException e) {
                System.err.println("InterruptedException: " + e.getMessage());
            } catch (Exception e) {
                System.err.println("Exception: " + e.getMessage());
            }


        } else {
            System.err.println("must input maxLine, tow file full path, and diff row number");
        }
    }

    public static void generateFile(Long maxLine, String fileFullPath, String snappyFullPath, List<Long> collectionOfDiff) throws IOException {
        Roaring64Bitmap rb = new Roaring64Bitmap();

        try (DataOutputStream out = new DataOutputStream(new FileOutputStream(fileFullPath));
             FileChannel targetChannel = FileChannel.open(Paths.get(snappyFullPath),
                StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {

            //ByteBuffer buffer = ByteBuffer.allocate(8192);

            if (collectionOfDiff != null && !collectionOfDiff.isEmpty()) {
                for (long k = 0; k < maxLine; k++) {
                    String datum = "aaaaaaaaaaaaaaaaaaaaaaaaa\n" + k;
                    if (collectionOfDiff.contains(k)) {
                        datum = "bbbbbbbbbbbbbbbbbbbbbbbbb\n" + k;
                    }
                    byte[] compressed = Snappy.compress(datum.getBytes(StandardCharsets.UTF_8));
                    targetChannel.write(ByteBuffer.wrap(compressed));
                    long crcId = MurmurHashUtil.getLongFromString(datum);
                    rb.add(crcId);
                }
            } else {
                for (long k = 0; k < maxLine; k++) {
                    String datum = "aaaaaaaaaaaaaaaaaaaaaaaaa\n" + k;
                    byte[] compressed = Snappy.compress(datum.getBytes(StandardCharsets.UTF_8));
                    targetChannel.write(ByteBuffer.wrap(compressed));
                    long crcId = MurmurHashUtil.getLongFromString(datum);
                    rb.add(crcId);
                }
            }

            rb.serialize(out);
        }
    }


    public static long getCost(long start) {
        return (System.currentTimeMillis() - start) / 1000;
    }
}